package com.jacquesh;

import com.jacquesh.model.CoffeeShop;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.RichCoFlatMapFunction;
import org.apache.flink.streaming.api.functions.co.RichCoMapFunction;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;

public class FlinkDemo0008 {

    public static void main(String[] args) throws Exception {
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");
        StreamExecutionEnvironment environment =
                StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> stream1 = environment
                .socketTextStream("192.168.88.180", 8888);
        DataStreamSource<String> stream2 = environment
                .socketTextStream("192.168.88.181", 8888);
        ConnectedStreams connectedStreams = stream1.connect(stream2);
        DataStream dataStream = connectedStreams.flatMap(new RichCoFlatMapFunction<String,String, CoffeeShop>() {

            @Override
            public void flatMap1(String s, Collector<CoffeeShop> collector)
                    throws Exception {
             String[] sp=s.split(",");
                collector.collect(CoffeeShop.builder()
                        .id(Long.parseLong(sp[0]))
                        .date(format.parse(sp[2]))
                        .sales(Integer.parseInt(sp[3]))
                        .build());
            }
            @Override
            public void flatMap2(String s, Collector<CoffeeShop> collector)
                    throws Exception {
                String[] sp=s.split(",");
                collector.collect(CoffeeShop.builder()
                        .id(Long.parseLong(sp[0]))
                        .date(format.parse(sp[1]))
                        .sales(Integer.parseInt(sp[2]))
                        .build());
            }
        });
        dataStream.print(">>>>>>>").setParallelism(1);
        environment.execute("connect demo");
    }
    }

